package com.aisun.flinkservice.tofile;

import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.accumulators.LongCounter;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.FileSystem;

/**
 * @Author: LiuYuanQi
 * @Date: 2020/4/29 10:08
 * 生成文件到指定文件
 */
public class List2File {

    public static void main(String[] args) throws Exception {

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSource<String> data = env.fromElements("hadoop","spark","flink","pyspark","storm");

        DataSet<String> info = data.map(new RichMapFunction<String, String>() {

            LongCounter counter = new LongCounter();

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);

                // step1: 初始化计数器
                getRuntimeContext().addAccumulator("ele-counts-java", counter);
            }

            @Override
            public String map(String value)  {
                // step2: 计数器迭代
                counter.add(1);
                return value;
            }
        });
        String filePath = "f://data";//    存储文件的目录
        info.writeAsText(filePath, FileSystem.WriteMode.OVERWRITE).setParallelism(2);
        JobExecutionResult jobResult = env.execute("CounterApp");

        // step3: 获取计数器
        long num = jobResult.getAccumulatorResult("ele-counts-java");
        System.out.println("num = [" + num + "]");
    }

}
